package drds.global_transaction.api.rpc.rpc_server;

import drds.global_transaction.api.api.RpcContext;
import drds.global_transaction.api.message_codec.Version;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.resource_manager.RegisterResourceManagerRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.register_manager.transaction_manager.RegisterTransactionManagerRequestMessage;
import drds.global_transaction.api.rpc.rpc_client.Role;
import drds.global_transaction.common.Constants;
import io.netty.channel.Channel;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;

import java.net.SocketAddress;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

@Slf4j
public class ChannelManager
{
    //public  static 	ThreadLocal<Channel> channelThreadLocal=new ThreadLocal<Channel>();
	/**
	 * channel RpcContext
	 */
	private static final ConcurrentMap<Channel, RpcContext> channelToRpcContextMap = new ConcurrentHashMap<Channel, RpcContext>();
	/**
	 * ResourceId+ApplicationId+ClientIp+ClientPort RpcContext
	 */
	private static final ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>>> //
	resourceManagerClientResourceIdToApplicationIdToClientIpToClientPortToRpcContextMapMapMapMap = new ConcurrentHashMap<String, ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>>>();

	/**
	 * ClientId<ip+applicationId> ClientPort RpcContext
	 */
	private static final ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> //
	transactionManagerClientClientIdToClientPortToRpcContextMapMap = new ConcurrentHashMap<String, ConcurrentMap<Integer, RpcContext>>();

	/**
	 *
	 * @param role                    #important
	 * @param version
	 * @param applicationId           #important
	 * @param transactionServiceGroup
	 * @param resourceIds
	 * @param channel                 #important
	 * @return
	 */
	private static RpcContext buildRpcContext(Role role, String version, String applicationId, String transactionServiceGroup, String resourceIds, Channel channel)
	{
		RpcContext rpcContext = new RpcContext();
		rpcContext.setRole(role);
		rpcContext.setVersion(version);
		rpcContext.setApplicationId(applicationId);
		rpcContext.setTransactionServiceGroup(transactionServiceGroup);
		rpcContext.addResourceId(resourceIdsToSet(resourceIds));
		rpcContext.setChannel(channel);
		return rpcContext;
	}

	private static String getAddress(Channel channel)
	{
		SocketAddress socketAddress = channel.remoteAddress();
		String address = socketAddress.toString();
		if (socketAddress.toString().indexOf(Constants.endpoint_begin_char) == 0)
		{
			address = socketAddress.toString().substring(Constants.endpoint_begin_char.length());
		}
		return address;
	}

	/**
	 * 根据channel获取端口相同的可用的Channel
	 */
	public static Channel getChannel(Channel channel)
	{
		if (channel.isActive())
		{
			return channel;
		}
		RpcContext rpcContext = getRpcContext(channel);
		if (null == rpcContext)
		{
			log.error("rpcContext is null,channel:" + channel + ",active:" + channel.isActive());
			return null;
		}
		if (rpcContext.getChannel().isActive())
		{
			// recheck
			return rpcContext.getChannel();
		}
		Integer clientPort = getClientPort(channel);
		Role role = rpcContext.getRole();
		if (role == Role.transaction_manager_client)
		{
			String clientId = rpcContext.getApplicationId() + Constants.client_id_split_char + getClientIp(channel);
			if (!transactionManagerClientClientIdToClientPortToRpcContextMapMap.containsKey(clientId))
			{
				return null;
			}
			ConcurrentMap<Integer, RpcContext> clientPortToRpcContextMap = transactionManagerClientClientIdToClientPortToRpcContextMapMap.get(clientId);
			return getChannel(clientPortToRpcContextMap, clientPort);
		} else if (role == Role.resource_manager_client)
		{
			for (Map<Integer, RpcContext> clientPortToRpcContextMap : rpcContext.getResourceIdToClientPortToRpcContextMapMap().values())
			{
				Channel $channel = getChannel(clientPortToRpcContextMap, clientPort);
				if (null != $channel)
				{
					return $channel;
				}
			}
		}
		return null;

	}

	/**
	 * 需要排除当前端口和其他非有效的端口
	 */
	private static Channel getChannel(Map<Integer, RpcContext> clientPortToRpcContextMap, int exclusivePort)
	{
		if (null != clientPortToRpcContextMap && !clientPortToRpcContextMap.isEmpty())
		{
			for (ConcurrentMap.Entry<Integer, RpcContext> entry : clientPortToRpcContextMap.entrySet())
			{
				// 当前端口删除
				if (entry.getKey().intValue() == exclusivePort)
				{
					clientPortToRpcContextMap.remove(entry.getKey());
					continue;
				}
				// 非有效删除
				Channel channel = entry.getValue().getChannel();
				if (channel.isActive())
				{
					return channel;
				}
				clientPortToRpcContextMap.remove(entry.getKey());
			}
		}
		return null;
	}

	public static Channel getChannel(String resourceId, String applicationId, String clientIp)
	{
		Channel $channel = null;
		ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>> applicationIdToClientIpToClientPortToRpcContextMapMapMap = //
				resourceManagerClientResourceIdToApplicationIdToClientIpToClientPortToRpcContextMapMapMapMap.get(resourceId);
		if (null != applicationIdToClientIpToClientPortToRpcContextMapMapMap && !applicationIdToClientIpToClientPortToRpcContextMapMapMap.isEmpty())
		{
			ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> clientIpToClientPortToRpcContextMapMap = //
					applicationIdToClientIpToClientPortToRpcContextMapMapMap.get(applicationId);
			//
			if (null != clientIpToClientPortToRpcContextMapMap && !clientIpToClientPortToRpcContextMapMap.isEmpty())
			{
				ConcurrentMap<Integer, RpcContext> clientPortToRpcContextMap = clientIpToClientPortToRpcContextMapMap.get(clientIp);
				//
				if (null != clientPortToRpcContextMap && !clientPortToRpcContextMap.isEmpty())
				{
					for (ConcurrentMap.Entry<Integer, RpcContext> entry : clientPortToRpcContextMap.entrySet())
					{
						Channel channel = entry.getValue().getChannel();
						if (channel.isActive())
						{
							$channel = channel;
							break;
						}
						clientPortToRpcContextMap.remove(entry.getKey());
					}
				}
				if (null == $channel)
				{
					for (ConcurrentMap.Entry<String, ConcurrentMap<Integer, RpcContext>> entry : clientIpToClientPortToRpcContextMapMap.entrySet())
					{
						if (entry.getKey().equals(clientIp))
						{// ip not equals
							continue;
						}
						for (ConcurrentMap.Entry<Integer, RpcContext> entry1 : entry.getValue().entrySet())
						{
							Channel channel = entry1.getValue().getChannel();
							if (channel.isActive())
							{
								$channel = channel;
								break;
							}
							entry.getValue().remove(entry1.getKey());
						}
						if (null != $channel)
						{
							break;
						}
					}
				}
			}
		}
		if (null == $channel)
		{
			for (ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>> $applicationIdToClientIpToClientPortToRpcContextMapMapMap : //
			resourceManagerClientResourceIdToApplicationIdToClientIpToClientPortToRpcContextMapMapMapMap.values())
			{
				for (ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> clientIpToClientPortToRpcContextMapMap : //
				$applicationIdToClientIpToClientPortToRpcContextMapMapMap.values())
				{
					if (clientIpToClientPortToRpcContextMapMap.containsKey(clientIp))
					{// 相同ip
						for (ConcurrentMap.Entry<Integer, RpcContext> entry : clientIpToClientPortToRpcContextMapMap.get(clientIp).entrySet())
						{
							Channel channel = entry.getValue().getChannel();
							if (channel.isActive())
							{
								$channel = channel;
								break;
							}
							clientIpToClientPortToRpcContextMapMap.get(clientIp).remove(entry.getKey());
						}
					}
					if (null != $channel)
					{
						break;
					}
				}
				if (null != $channel)
				{
					break;
				}
			}
		}

		return $channel;

	}

	private static String getClientIp(Channel channel)
	{
		String address = getAddress(channel);
		String clientIp = address;
		if (clientIp.contains(Constants.ip_port_split_char))
		{
			clientIp = clientIp.substring(0, clientIp.lastIndexOf(Constants.ip_port_split_char));
		}
		return clientIp;
	}

	private static Integer getClientPort(Channel channel)
	{
		String address = getAddress(channel);
		Integer port = 0;
		try
		{
			if (address.contains(Constants.ip_port_split_char))
			{
				port = Integer.parseInt(address.substring(address.lastIndexOf(Constants.ip_port_split_char) + 1));
			}
		} catch (NumberFormatException exx)
		{
			log.error(exx.getMessage());
		}
		return port;
	}

	public static RpcContext getRpcContext(Channel channel)
	{
		return channelToRpcContextMap.get(channel);
	}

	public static boolean isRegistered(Channel channel)
	{
		boolean isRegistered=channelToRpcContextMap.containsKey(channel);
		if(drds.common.Constants.developMode) {
			String logString="";
			for (Channel $channel: channelToRpcContextMap.keySet()) {
				logString="channel:" + $channel.remoteAddress().toString()+",id:"+$channel.id().asLongText() + ",isActive:" + $channel.isActive();
				log.debug(logString);
				//

				//channelThreadLocal.set($channel);
				//
			}
			//
			logString="channel:" + channel.remoteAddress().toString()+",id:"+channel.id().asLongText() + ",isRegistered:" + isRegistered;
			log.debug(logString);
		}
		return false;
	}

	public static void registerResourceManagerClientChannel(RegisterResourceManagerRequestMessage registerResourceManagerRequestMessage, Channel channel)
	{
		Version.checkVersion(registerResourceManagerRequestMessage.getVersion());
		Set<String> resourceIdSet = resourceIdsToSet(registerResourceManagerRequestMessage.getResourceIds());
		RpcContext rpcContext;
		if (!channelToRpcContextMap.containsKey(channel))
		{
			rpcContext = buildRpcContext(Role.resource_manager_client, //
					registerResourceManagerRequestMessage.getVersion(), //
					registerResourceManagerRequestMessage.getApplicationId(), //
					registerResourceManagerRequestMessage.getTransactionServiceGroup(), //
					registerResourceManagerRequestMessage.getResourceIds(), //
					channel);//
			rpcContext.map$channelToRpcContextMap(channelToRpcContextMap);
		} else
		{
			rpcContext = channelToRpcContextMap.get(channel);
			rpcContext.addResourceId(resourceIdSet);
		}
		if (null == resourceIdSet || resourceIdSet.isEmpty())
		{
			return;
		}
		for (String resourceId : resourceIdSet)
		{
			resourceManagerClientResourceIdToApplicationIdToClientIpToClientPortToRpcContextMapMapMapMap.putIfAbsent(resourceId, new ConcurrentHashMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>>());
			ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>> applicationIdToClientIpToClientPortToRpcContextMapMapMap = //
					resourceManagerClientResourceIdToApplicationIdToClientIpToClientPortToRpcContextMapMapMapMap.get(resourceId);
			//
			applicationIdToClientIpToClientPortToRpcContextMapMapMap.putIfAbsent(registerResourceManagerRequestMessage.getApplicationId(), new ConcurrentHashMap<String, ConcurrentMap<Integer, RpcContext>>());
			ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> clientIpToClientPortToRpcContextMapMap = //
					applicationIdToClientIpToClientPortToRpcContextMapMapMap.get(registerResourceManagerRequestMessage.getApplicationId());
			//
			String clientIp = getClientIp(channel);
			clientIpToClientPortToRpcContextMapMap.putIfAbsent(clientIp, new ConcurrentHashMap<Integer, RpcContext>());
			//
			ConcurrentMap<Integer, RpcContext> clientPortToRpcContextMap = clientIpToClientPortToRpcContextMapMap.get(clientIp);
			rpcContext.map$resourceIdToClientPortToRpcContextMapMap(resourceId, clientPortToRpcContextMap);
			//
			updateClientPortToRpcContextMap(resourceId, clientIp, registerResourceManagerRequestMessage.getApplicationId());
		}

	}

	public static void registerTransactionManagerClientChannel(RegisterTransactionManagerRequestMessage registerTransactionManagerRequest, Channel channel)
	{
		Version.checkVersion(registerTransactionManagerRequest.getVersion());
		RpcContext rpcContext = buildRpcContext(//
				Role.transaction_manager_client, //
				registerTransactionManagerRequest.getVersion(), //
				registerTransactionManagerRequest.getApplicationId(), //
				registerTransactionManagerRequest.getTransactionServiceGroup(), //
				null, //
				channel);//
		rpcContext.map$channelToRpcContextMap(channelToRpcContextMap);
		//
		String clientId = rpcContext.getApplicationId() + Constants.client_id_split_char + getClientIp(channel);
		//
		transactionManagerClientClientIdToClientPortToRpcContextMapMap.putIfAbsent(clientId, new ConcurrentHashMap<Integer, RpcContext>());
		ConcurrentMap<Integer, RpcContext> clientPortToRpcContextMap = transactionManagerClientClientIdToClientPortToRpcContextMapMap.get(clientId);//
		rpcContext.map$clientPortToRpcContextMap(clientPortToRpcContextMap);
	}

	public static void releaseRpcContext(@NonNull Channel channel)
	{
		if (channelToRpcContextMap.containsKey(channel))
		{
			RpcContext rpcContext = getRpcContext(channel);
			if (rpcContext != null)
			{
				rpcContext.release();
			}

		}
	}

	private static Set<String> resourceIdsToSet(String resourceIds)
	{
		if (StringUtils.isEmpty(resourceIds))
		{
			return null;
		}
		Set<String> set = new HashSet<String>();
		for (String string : resourceIds.split(Constants.resourceIds_split_char))
		{
			set.add(string);
		}
		return set;
	}

	private static void updateClientPortToRpcContextMap(String resourceId, String clientIp, String applicationId)
	{
		// resourceId applicationId clientIp
		ConcurrentMap<Integer, RpcContext> clientPortToRpcContextMap = resourceManagerClientResourceIdToApplicationIdToClientIpToClientPortToRpcContextMapMapMapMap.//
				get(resourceId).//
				get(applicationId).//
				get(clientIp);//
		for (ConcurrentMap.Entry<String, ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>>> entry : //
		resourceManagerClientResourceIdToApplicationIdToClientIpToClientPortToRpcContextMapMapMapMap.entrySet())
		{//
			if (entry.getKey().equals(resourceId))
			{
				continue;
			}
			ConcurrentMap<String, ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>>> applicationIdToClientIpToClientPortToRpcContextMapMapMap = entry.getValue();
			if (!applicationIdToClientIpToClientPortToRpcContextMapMapMap.containsKey(applicationId))
			{
				continue;
			}
			ConcurrentMap<String, ConcurrentMap<Integer, RpcContext>> clientIpToClientPortToRpcContextMapMap = applicationIdToClientIpToClientPortToRpcContextMapMapMap.get(applicationId);
			if (!clientIpToClientPortToRpcContextMapMap.containsKey(clientIp))
			{
				continue;
			}
			// the same ip
			ConcurrentMap<Integer, RpcContext> $clientPortToRpcContextMap = clientIpToClientPortToRpcContextMapMap.get(clientIp);
			for (ConcurrentMap.Entry<Integer, RpcContext> $entry : $clientPortToRpcContextMap.entrySet())
			{
				Integer clientPort = $entry.getKey();
				if (!clientPortToRpcContextMap.containsKey(clientPort))
				{
					RpcContext rpcContext = $entry.getValue();
					clientPortToRpcContextMap.put(clientPort, rpcContext);
					rpcContext.map$resourceIdToClientPortToRpcContextMapMap(resourceId, clientPort);
				}
			}
		}
	}
}
